package com.simple.datax.conn.impl;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.SimpleMessage;
import com.simple.datax.api.ConnectorException;

/** 
 * @Title: ActiveMQConnImpl.java 
 * @Package com.simple.datax.conn.impl 
 * @Description: 实现具体的ActiveMQ连接操作，包括连接，接收，发送等。 
 * @author Jin Yi 
 * @date 2012-12-30
 * @version V1.0 
 */
public class ActiveMQConnImpl extends MQBaseConnector {
	private static final Logger logger = LoggerFactory
			.getLogger(ActiveMQConnImpl.class);
	public static final String TYPE = "ActiveMQ";
	private String username;
	private String password;
	private String brokeURL;
	
	private Connection connection;
	private Session session;
	private String subject;
	private Destination destination = null;
	private MessageConsumer consumer = null;
	private MessageProducer producer = null;
	
	/**
	 * 构造器
	 * @param connPara
	 * @throws ConnectorException
	 */
	public ActiveMQConnImpl(String connPara) throws ConnectorException {
		super(connPara);
		this.connPara = connPara;
		parseConnPara();
		createMQ();
	}
	
	/**
	 * 创建并初始化ActiveMQ对象
	 * @throws ConnectorException
	 */
	private void createMQ() throws ConnectorException {
		try {
			ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(     
					username, password, brokeURL);   
			//连接工厂创建一个jms connection
			connection = connectionFactory.createConnection();     
			//是生产和消费的一个单线程上下文。会话用于创建消息的生产者，消费者和消息。会话提供了一个事务性的上下文。  
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			connection.start();
		} catch (JMSException e) {
			throw new ConnectorException("Creating ActiveMQ failed - brokeURL" + brokeURL, e);
		}
	}
	
	/**
	 * 解析连接参数
	 * @throws ConnectorException 
	 */
	private void parseConnPara() throws ConnectorException {
		//connPara = "URL$USER$PASSWORD$SUBJECT";
		String[] tmpStr = splitConnPara(connPara);
		if (tmpStr.length==4) {
			brokeURL = tmpStr[0];
			username = tmpStr[1];
			password = tmpStr[2];
			subject = tmpStr[3];
		} else {
			throw new ConnectorException(
					"Parsing ActiveMQ connector parameters failed. Please check parameter - "
							+ connPara);
		}
	}
	
	@Override
	public String getMessage() throws ConnectorException {
		String ret = null;
		prepareReceiveQueue();
		try {
			Message message = consumer.receive(RECV_NO_MESSAGE_TIMOUT);
			if (message instanceof TextMessage) {
				TextMessage textMsg = (TextMessage) message;
				if (textMsg != null) {
					ret = textMsg.getText();
					ret = getInputString(ret, getEncoding());
				}
			}
			if (message instanceof BytesMessage) {
				BytesMessage bytesMsg = (BytesMessage) message;
				if (bytesMsg != null) {
					ret = getMessage(bytesMsg, getEncoding());
				}
			}
		} catch (Exception ex) {
			throw new ConnectorException("Receive message from ActiveMQ failed.", ex);
		}
		return ret;
	}

	/**
	 * 为发送消息做准备
	 * @throws ConnectorException
	 */
	private void prepareSendQueue() throws ConnectorException{
		try {
			if (destination == null || producer == null) {
				destination = null;
				destination = session.createQueue(subject);
				producer = null;
				producer = session.createProducer(destination);
//				producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//				connection.start();
			}
		} catch (Exception e) {
			throw new ConnectorException("ActiveMQ producer initialization failed", e);
		}
	}
	
	/**
	 * 为接收消息做准备
	 * @throws ConnectorException
	 */
	private void prepareReceiveQueue() throws ConnectorException{
		try {
			if (destination == null || consumer == null) {
				destination = null;
				destination = session.createQueue(subject);
				consumer = null;
				consumer = session.createConsumer(destination);
//				consumer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//				connection.start();
			}
		} catch (Exception e) {
			throw new ConnectorException("ActiveMQ producer initialization failed", e);
		}
	}

	@Override
	public String getConnType() {
		return TYPE;
	}

	@Override
	public void setMessageListener(MessageListener listener) throws ConnectorException {
		try {
			prepareReceiveQueue();
			consumer.setMessageListener(listener);
		} catch (Exception e) {
			throw new ConnectorException("Setting ActiveMQ listener failed.", e);
		}
	}
	
	@Override
	protected void prepareSend() throws ConnectorException {
		prepareSendQueue();
	}

	@Override
	protected void internalSend(SimpleMessage message) throws ConnectorException {
		try {
			BytesMessage msg = session.createBytesMessage();
			msg.writeBytes(getOutputBytes(message.getContent(), getEncoding()));
//			String content = getOutputString(message.getContent(), getEncoding());
//			TextMessage msg = session.createTextMessage(content);
			producer.send(msg);
		} catch (JMSException ex) {
			throw new ConnectorException("ActiveMQ internalSend failed", ex, true);
		} catch (Exception ex) {
			throw new ConnectorException("ActiveMQ internalSend failed", ex);
		}
	}

	@Override
	public void disconnect() {
		if (producer != null) {
			try {
				producer.close();
			} catch (JMSException e) {
				logger.warn("ActiveMQ producer disconnect error", e);
			}
			producer = null;
		}
		
		if (consumer != null) {
			try {
				consumer.close();
			} catch (JMSException e) {
				logger.warn("ActiveMQ consumer disconnect error", e);
			}
			consumer = null;
		}
		
		if (session != null) {
			try {
				session.close();
			} catch (JMSException e) {
				logger.warn("ActiveMQ session disconnect error", e);
			}
			session = null;
		}
		
		if (connection != null) {
			try {
				connection.stop();
			} catch (Exception e) {
				logger.warn("ActiveMQ connection disconnect error", e);
			}
			connection = null;
		}
	}

	@Override
	protected void prepareReceive() throws ConnectorException {
		prepareReceiveQueue();
	}

}
